<?php
/**
 * Jcer Client API for Php 
 * 
 * @author quanwei
 *
 */
class JcerClient extends Socket {
	const TASKID_ECHO = 1;
	const TASKID_HTTP = 2;
	const TASKID_SQL = 3;
	
	public $packet;
	function JcerClient() {
		$this->packet = new Packet ();
	}
	public function addTask($taskId, $taskNum, $stream) {
		$this->packet->packTask ( $taskId, $taskNum, $stream );
	}
	public function getResponse() {
		$this->packet->endPack ();
		$stream = $this->packet->stream;
		$this->packet = new Packet ();
		$this->connect ();
		$this->write ( $stream );
		
		$stream = $this->read ( Packet::getHeaderLen () );
		$packet = new Packet ();
		$packet->unpackHeader ( $stream );
		$stream = $this->read ( $packet->bodyLen );
		$packet->unpackBody ( $stream );
		return $packet->data;
	}
}
class SqlJcerClient extends JcerClient {
	public function addSql($sourceKey, $taskNum, $sql) {
		$this->addTask ( self::TASKID_SQL, $taskNum, $sourceKey . ":" . $sql );
	}
	public function getResponse() {
		$response = parent::getResponse ();
		foreach ( $response as $taskNum => $value ) {
			$response [$taskNum] = json_decode ( $value, true );
		}
		return $response;
	
	}
}
class HttpJcerClient extends JcerClient {
	public function addUrl($taskNum, $url) {
		$this->addTask ( self::TASKID_HTTP, $taskNum, $url );
	}
}
class Packet {
	public $version = "jcer1.0.0";
	public $bodyLen;
	public $cmd = 1;
	public $magicNum = 0xff;
	public $taskCount = 0;
	
	public $header;
	public $body;
	public $stream;
	
	public $data;
	
	public static function getHeaderLen() {
		return 22;
	}
	
	public function packHeader() {
		$this->header = "";
		$this->header .= pack ( "a10", $this->version );
		$this->header .= pack ( "N", $this->bodyLen );
		$this->header .= pack ( "N", $this->cmd );
		$this->header .= pack ( "N", $this->magicNum );
	}
	public function packTask($taskId, $taskNum, $stream) {
		$this->body .= pack ( "N", 4 + 4 + strlen ( $stream ) );
		$this->body .= pack ( "N", $taskId );
		$this->body .= pack ( "N", $taskNum );
		$this->body .= pack ( "a" . strlen ( $stream ), $stream );
		$this->taskCount ++;
	}
	public function unpackHeader($stream) {
		$tmp = unpack ( "a10", $stream );
		$this->version = $tmp [1];
		$stream = substr ( $stream, 10 );
		
		$tmp = unpack ( "N", $stream );
		$this->bodyLen = $tmp [1];
		$stream = substr ( $stream, 4 );
		
		$tmp = unpack ( "N", $stream );
		$this->cmd = $tmp [1];
		$stream = substr ( $stream, 4 );
		
		$tmp = unpack ( "N", $stream );
		$this->magicNum = $tmp [1];
		$stream = substr ( $stream, 4 );
	}
	public function unpackBody($stream) {
		$this->data = array ();
		$tmp = unpack ( "N", $stream );
		$count = $tmp [1];
		$stream = substr ( $stream, 4 );
		for($i = 0; $i < $count; $i ++) {
			$row = array ();
			
			$tmp = unpack ( "N", $stream );
			$len = $tmp [1];
			$stream = substr ( $stream, 4 );
			
			$tmp = unpack ( "N", $stream );
			$row ['taskId'] = $tmp [1];
			$stream = substr ( $stream, 4 );
			
			$tmp = unpack ( "N", $stream );
			$row ['taskNum'] = $tmp [1];
			$stream = substr ( $stream, 4 );
			$tmp = unpack ( "a" . ($len - 8), $stream );
			$row ['response'] = $tmp [1];
			$stream = substr ( $stream, $len - 8 );
			
			$this->data [$row ['taskNum']] = $row ['response'];
		}
	}
	public function endPack() {
		$this->body = pack ( "N", $this->taskCount ) . $this->body;
		$this->bodyLen = strlen ( $this->body );
		$this->packHeader ();
		$this->stream = $this->header . $this->body;
	}

}
class Socket {
	protected $_servers = array ();
	protected $_fd;
	
	protected $_readTimeout;
	protected $_writeTimeout;
	
	const LEVEL_ERROR = 1;
	const LEVEL_INFO = 2;
	const LEVEL_DEBUG = 3;
	const LEVEL_WARNING = 4;
	const LEVEL_TRACE = 5;
	
	/**
	 *
	 * @return the $_readTimeout
	 */
	public function getReadTimeout() {
		return $this->_readTimeout;
	}
	
	/**
	 *
	 * @return the $_writeTimeout
	 */
	public function getWriteTimeout() {
		return $this->_writeTimeout;
	}
	
	/**
	 *
	 * @param $_readTimeout field_type       	
	 */
	public function setReadTimeout($readTimeout) {
		$this->_readTimeout = $readTimeout;
	}
	
	/**
	 *
	 * @param $_writeTimeout field_type       	
	 */
	public function setWriteTimeout($writeTimeout) {
		$this->_writeTimeout = $writeTimeout;
	}
	
	protected function _write($fd, $stream) {
		if (! is_resource ( $fd )) {
			throw new Exception ( "connect not created" );
		}
		
		$remain = strlen ( $stream );
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$stime = $sec + $usec;
		
		while ( $remain > 0 ) {
			if ($this->_writeTimeout > 0) {
				// 超时判断
				list ( $sec, $usec ) = explode ( " ", microtime () );
				$etime = $sec + $usec;
				if (($etime - $stime) * 1000 > $this->_writeTimeout) {
					throw new Exception ( "syserror:write timeout,readtimeou:" . ($etime - $stime) );
				}
			}
			
			$len = @socket_write ( $fd, $stream, $remain );
			if ($len === false) {
				$errno = socket_last_error ();
				if ($errno == SOCKET_EINPROGRESS || $errno == SOCKET_EAGAIN) {
					continue;
				}
				// 不是EINPROGRESS，表连接失败，抛出异常
				$errmsg = socket_strerror ( $errno );
				list ( $sec, $usec ) = explode ( " ", microtime () );
				$etime = $sec + $usec;
				throw new Exception ( "socket_send error:[$errno:$errmsg],timecost=" . ($etime - $stime) );
			}
			$remain -= $len;
		}
	}
	
	protected function _read($fd, $length) {
		if (! is_resource ( $fd )) {
			throw new Exception ( "connect not created" );
		}
		
		$data = "";
		$remain = $length;
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$stime = $sec + $usec;
		
		while ( $remain > 0 ) {
			if ($this->_readTimeout > 0) {
				// 超时判断
				list ( $sec, $usec ) = explode ( " ", microtime () );
				$etime = $sec + $usec;
				if (($etime - $stime) * 1000 > $this->_readTimeout) {
					throw new Exception ( "syserror:read timeout,readtimeou:" . ($etime - $stime) );
				}
			}
			
			$buf = @socket_read ( $fd, $remain, PHP_BINARY_READ );
			if ($buf === false) {
				$errno = socket_last_error ();
				if ($errno == SOCKET_EINPROGRESS || $errno == SOCKET_EAGAIN) {
					continue;
				}
				// 不是EINPROGRESS，表连接失败，抛出异常
				$errmsg = socket_strerror ( $errno );
				list ( $sec, $usec ) = explode ( " ", microtime () );
				$etime = $sec + $usec;
				throw new Exception ( "socket_recv error:[$errno:$errmsg],timecost:" . ($etime - $stime) );
			}
			$len = strlen ( $buf );
			$remain -= $len;
			$data .= $buf;
			
			if ($len == 0 && $remain > 0) {
				throw new Exception ( "socket_recv error:server closed" );
			}
		}
		return $data;
	}
	
	/**
	 * 最多读取length个字节
	 * 终止条件：读到了字节且无数据可读 || 超时 || 出错
	 *
	 * @param $fd int       	
	 * @param $length int       	
	 * @throws Exception
	 */
	protected function _tryRead($fd, $length) {
		if (! is_resource ( $fd )) {
			throw new Exception ( "connect not created" );
		}
		
		$data = "";
		$remain = $length;
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$stime = $sec + $usec;
		
		while ( $remain > 0 ) {
			if ($this->_readTimeout > 0) {
				// 超时判断
				list ( $sec, $usec ) = explode ( " ", microtime () );
				$etime = $sec + $usec;
				if (($etime - $stime) * 1000 > $this->_readTimeout) {
					throw new Exception ( "syserror:read timeout,readtimeou:" . ($etime - $stime) );
				}
			}
			
			$buf = @socket_read ( $fd, $remain, PHP_BINARY_READ );
			if ($buf === false) {
				$errno = socket_last_error ();
				if ($errno == SOCKET_EINPROGRESS || $errno == SOCKET_EAGAIN) {
					continue;
				}
				// 不是EINPROGRESS，表连接失败，抛出异常
				$errmsg = socket_strerror ( $errno );
				list ( $sec, $usec ) = explode ( " ", microtime () );
				$etime = $sec + $usec;
				throw new Exception ( "socket_recv error:[$errno:$errmsg],timecost:" . ($etime - $stime) );
			}
			$len = strlen ( $buf );
			$remain -= $len;
			$data .= $buf;
			
			if ($len == 0 && $remain > 0) {
				throw new Exception ( "socket_recv error:server closed" );
			}
			
			if ($len > 0) {
				break;
			}
		}
		return $data;
	}
	
	/**
	 *
	 * @return the $_servers
	 */
	public function getServers() {
		return $this->_servers;
	}
	
	/**
	 *
	 * @param $_servers field_type       	
	 */
	public function setServers($servers) {
		$this->_servers = $servers;
	}
	
	function isAlive() {
		return is_resource ( $this - _fd );
	}
	
	function connect($force = false) {
		if (! $force && is_resource ( $this->_fd )) {
			// 如果非强制连接，则复用之前的连接
			return;
		}
		
		$servers = $this->_servers;
		if (! is_array ( $servers )) {
			throw new Exception ( "server is not define" );
		}
		
		shuffle ( $servers );
		foreach ( $servers as $key => $server ) {
			try {
				$this->_connect ( $server );
				break;
			} catch ( Exception $ex ) {
				// 连过的
				$logmsg = sprintf ( "connect server %s fail,exception:'%s'", $server ['host'], $ex->getMessage () );
				$this->log ( $logmsg, self::LEVEL_WARNING, __CLASS__ );
			}
		}
		
		// check是否链接成功
		if (! is_resource ( $this->_fd )) {
			$logmsg = sprintf ( "connect all server fail,exception:'%s'", $ex->getMessage () );
			$this->log ( $logmsg, self::LEVEL_ERROR, __CLASS__ );
			throw new Exception ( "connect all server fail" );
		}
		
		$logmsg = sprintf ( "connect server %s succ", $server ['host'] );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
	
	}
	
	private function _connect($server) {
		// 申请句柄
		$fd = @socket_create ( AF_INET, SOCK_STREAM, SOL_TCP );
		if (! $fd) {
			$errno = socket_last_error ();
			$errmsg = socket_strerror ( $errno );
			throw new Exception ( "socket_create error:[$errno:$errmsg]" );
		}
		
		// 连接
		if (! @socket_connect ( $fd, $server ['host'], $server ['port'] )) {
			$errno = socket_last_error ();
			$errmsg = socket_strerror ( $errno );
			throw new Exception ( "socket_create error:[$errno:$errmsg]" );
		}
		// set nonblock
		socket_set_nonblock ( $fd );
		$this->_fd = $fd;
	}
	
	function write($stream) {
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$stime = $sec + $usec;
		$logmsg = sprintf ( "write start at %s", $stime );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
		
		$this->_write ( $this->_fd, $stream );
		
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$etime = $sec + $usec;
		$logmsg = sprintf ( "write end at %s", $etime );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
	}
	
	function read($length) {
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$stime = $sec + $usec;
		$logmsg = sprintf ( "read start at %s", $stime );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
		
		return $this->_read ( $this->_fd, $length );
		
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$etime = $sec + $usec;
		$logmsg = sprintf ( "read end at %s", $etime );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
	}
	
	function readLine() {
		$line = "";
		$char = "";
		
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$stime = $sec + $usec;
		$logmsg = sprintf ( "readline start at %s", $stime );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
		
		while ( $char !== "\n" ) {
			$char = $this->_read ( $this->_fd, 1 );
			$line .= $char;
		}
		return $line;
		
		list ( $sec, $usec ) = explode ( " ", microtime () );
		$etime = $sec + $usec;
		$logmsg = sprintf ( "readline end at %s", $etime );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
	}
	public function tryRead($length) {
		return $this->_tryRead ( $this->_fd, $length );
	}
	
	function close() {
		$logmsg = sprintf ( "close socket,fd is %s", $this->_fd );
		$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
		
		if (! is_resource ( $this->_fd )) {
			$logmsg = sprintf ( "socket is not created" );
			$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
			return;
		}
		if (! @socket_close ( $this->_fd )) {
			$errno = socket_last_error ();
			$errmsg = socket_strerror ( $errno );
			$logmsg = sprintf ( "close conncet fail,errno=%s,errmsg=%s]", $errno, $errmsg );
			$this->log ( $logmsg, self::LEVEL_TRACE, __CLASS__ );
		}
		$this->_fd = null;
	}
	
	function log() {
	
	}
}